package org.mq.mymq.broker.core;

import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.model.*;
import org.mq.mymq.broker.utils.LogFileNameUtil;
import org.mq.mymq.broker.utils.PutMessageLock;
import org.mq.mymq.broker.utils.UnfairReentrantLock;
import org.mq.mymq.common.constants.BrokerConstants;
import org.mq.mymq.common.dto.MessageDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName MMapFileModel
 * @Description 最基础的mmap模型
 * @Author jiarong_ye
 * @Date 2025/6/15 15:36
 * @Version 1.0
 */
public class CommitLogMMapFileModel {
    private static final Logger log = LoggerFactory.getLogger(CommitLogMMapFileModel.class);
    private File file;
    private MappedByteBuffer mappedByteBuffer;
    private ByteBuffer readBuffer;
    private FileChannel fileChannel;
    private String topic;

    private PutMessageLock putMessageLock;


    class CommitLogFilePath{
        private String fileName;
        private String filePath;

        public CommitLogFilePath(String fileName, String filePath) {
            this.fileName = fileName;
            this.filePath = filePath;
        }

        public String getFileName() {
            return fileName;
        }

        public void setFileName(String fileName) {
            this.fileName = fileName;
        }

        public String getFilePath() {
            return filePath;
        }

        public void setFilePath(String filePath) {
            this.filePath = filePath;
        }
    }

    /**
     * 指定offset做文件的映射
     *
     * @param topicName 消息主题
     * @param startOffset 开始映射的offset
     * @param mappedSize  映射文件大小 单位：byte
     */
    public void loadFileInMMap(String topicName, int startOffset, int mappedSize) throws IOException {
        this.topic = topicName;
        String filePath = getLatestCommitLogFile(topicName);
        this.doMMap(filePath, startOffset, mappedSize);
        putMessageLock = new UnfairReentrantLock();
    }

    /**
     * 执行mmap步骤
     *
     * @param filePath 待映射文件路径
     * @param startOffset 开始映射的offset
     * @param mappedSize 映射文件大小 单位：byte
     * @throws IOException
     */
    private void doMMap(String filePath, int startOffset, int mappedSize) throws IOException {
        file = new File(filePath);
        if (!file.exists() || file.isDirectory()) {
            throw new FileNotFoundException("filePath is " + filePath + " inValid");
        }
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topic);
        this.fileChannel = new RandomAccessFile(file, "rw").getChannel();
        this.mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, startOffset, mappedSize);
        this.readBuffer = mappedByteBuffer.slice();
        this.mappedByteBuffer.position(myMqTopicModel.getCommitLog().getOffset().get());
    }

    /**
     * 获取最新commitLog文件
     *
     * @param topicName 主题
     * @return
     * @throws IOException
     */
    private String getLatestCommitLogFile(String topicName) throws IOException {
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topicName);
        if (myMqTopicModel == null) {
            throw new IllegalArgumentException("topicName is " + topicName + " inValid");
        }
        CommitLogModel commitLog = myMqTopicModel.getCommitLog();
        int diff = commitLog.countDiff();
        String filePath = "";
        if (diff == 0) {
            // 当前文件已经写满
            CommitLogFilePath commitLogFilePath = this.createNewCommitLogFile(topicName, commitLog);
            return commitLogFilePath.getFileName();
        } else if (diff > 0) {
            filePath = LogFileNameUtil.buildCommitLogFilePath(topicName, commitLog.getFileName());
        }
        return filePath;
    }

    /**
     * 创建新的CommitLog文件
     *
     * @param topicName CommitLog主题名称
     * @param commitLog 旧的CommitLog
     * @return 新CommitLog文件路径
     */
    private CommitLogFilePath createNewCommitLogFile(String topicName, CommitLogModel commitLog) {
        String newFileName = LogFileNameUtil.incrCommitLogName(commitLog.getFileName());
        String newFilePath = LogFileNameUtil.buildCommitLogFilePath(topicName, newFileName);
        File newCommitLogFile = new File(newFilePath);
        try {
            newCommitLogFile.createNewFile();
            log.info("创建了新的commitLog文件{}", newFilePath);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return new CommitLogFilePath(newFileName, newFilePath);
    }

    /**
     * 支持从文件的指定offset开始读取内容
     *
     * @param pos 读取位置
     * @param length 读取长度
     * @return
     */
    public byte[] readContent(int pos, int length) {
        byte[] content = new byte[length];
        ByteBuffer byteBuffer = readBuffer.slice();
        byteBuffer.position(pos);
        byteBuffer.get(content);
        return content;
    }


    /**
     * 写入内容后不强制刷盘（写入性能高，存在数据丢失风险）
     *
     * @param messageDTO
     * @return
     */
    public void writeContent(MessageDTO messageDTO) throws IOException {
        this.writeContent(messageDTO, false);
    }

    /**
     * 写入数据到磁盘当中 强制刷盘情况下写入性能相对差 数据丢失风险较低
     *
     * @param messageDTO
     * @return
     */
    public void writeContent(MessageDTO messageDTO, boolean force) throws IOException {
        // 定位到最新的commitLog文件中，记录当前文件是否已经写满，如果写满，则创建新的文件，并且创建新的mmap映射。
        // 如果当前文件没有写满，对content内容做一层封装，再判断写入是否会写满，如果不会则选择当前commitLog，如果会则创建新的文件并mmap
        // 定位到最新的commitLog文件后，写入
        // 定义一个对象专门管理各个topic的最新写入offset值，并且定时刷新到磁盘中
        // 写入数据，offset变更，如果是高并发场景，offset是不是会被多个线程访问？
        // offset使用原子类AtomicLong保证并发安全
        // 多线程写入内容使用锁保证并发安全，为了保证高性能，锁的选择很重要。
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topic);
        if (myMqTopicModel == null) {
            throw new IllegalArgumentException("topicName is " + topic + ", MQ topic inValid");
        }
         CommitLogModel commitLog = myMqTopicModel.getCommitLog();
        if (commitLog == null) {
            throw new IllegalArgumentException("topicName is " + topic + " MQ commitLog inValid");
        }

//        ByteBuffer byteBuffer = mappedByteBuffer.slice();
//        byteBuffer.position(111);
//        byteBuffer.put(content);
        try {
            putMessageLock.lock();
            CommitLogMessageModel commitLogMessageModel = new CommitLogMessageModel();
            commitLogMessageModel.setContent(messageDTO.getBody());
            this.checkCommitLogHasEnableSpace(commitLogMessageModel);
            //默认刷到page cache中
            byte[] content = commitLogMessageModel.getContent();
            mappedByteBuffer.put(content);
            this.dispatche(messageDTO, commitLog.getOffset().get());
            commitLog.getOffset().addAndGet(content.length);
            if (force) {
                //强制刷盘
                mappedByteBuffer.force();
            }
        }finally {
            putMessageLock.unlock();
        }
    }

    /**
     * 将ConsumerQueue文件写入队列
     *
     * @param messageDTO 分发内容
     * @param msgIndex     消息索引
     * @return
     */
    private void dispatche(MessageDTO messageDTO, int msgIndex) throws IOException {
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(topic);
        if (myMqTopicModel == null) {
            throw new IllegalArgumentException("topicName is " + topic + ", MQ topic inValid");
        }
        // 按照指定队列存放，没有指定队列id时随机存放
        int queueId;
        if (messageDTO.getQueueId() >= 0) {
            queueId = messageDTO.getQueueId();
        } else {
            int queueSize = myMqTopicModel.getQueueList().size();
            queueId = new Random().nextInt(queueSize);
        }
        // 分发消息（只包含索引信息）
        ConsumeQueueDetailModel consumeQueueDetailModel = new ConsumeQueueDetailModel();
        consumeQueueDetailModel.setCommitLogIndex(Integer.valueOf(myMqTopicModel.getCommitLog().getFileName()));
        consumeQueueDetailModel.setMsgIndex(msgIndex);
        consumeQueueDetailModel.setMsgLength(messageDTO.getBody().length);
        byte[] content = consumeQueueDetailModel.converToBytes();
        List<ConsumeQueueMMapFileModel> consumeQueueMMapFileModels = CommonCache.getConsumerQueueMMapFileModelManager().get(topic);
        ConsumeQueueMMapFileModel consumeQueueMMapFileModel = consumeQueueMMapFileModels.stream().
                filter(queueModel -> queueModel.getQueueId().equals(queueId)).findFirst().orElse(null);
        consumeQueueMMapFileModel.writeContent(content, true);
        // 刷新offset
        QueueModel queueModel = myMqTopicModel.getQueueList().get(queueId);
        queueModel.getLatestOffset().addAndGet(content.length);
    }

    private void checkCommitLogHasEnableSpace(CommitLogMessageModel commitLogMessageModel) throws IOException {
        MyMqTopicModel myMqTopicModel = CommonCache.getMyMqTopicModelsMap().get(this.topic);
        CommitLogModel commitLog = myMqTopicModel.getCommitLog();
        long writeAbleSize = commitLog.countDiff();
        if (!(writeAbleSize >= commitLogMessageModel.getSize())){
            CommitLogFilePath newCommitLogFile = this.createNewCommitLogFile(this.topic, commitLog);
            commitLog.setOffsetLimit(BrokerConstants.COMMITLOG_DEFAULT_SIZE);
            commitLog.setOffset(new AtomicInteger(0));
            commitLog.setFileName(newCommitLogFile.getFileName());
            this.doMMap(newCommitLogFile.getFilePath(), 0, BrokerConstants.COMMITLOG_DEFAULT_SIZE);
        }
    }


    /**
     * 释放mmap内存占用，主要通过将MappedByteBuffer中的attr引用清除
     */
    public void clean() {
        if (mappedByteBuffer == null || !mappedByteBuffer.isDirect() || mappedByteBuffer.capacity() == 0)
            return;
        invoke(invoke(viewed(mappedByteBuffer), "cleaner"), "clean");
    }

    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged(new PrivilegedAction<Object>() {
            public Object run() {
                try {
                    Method method = method(target, methodName, args);
                    method.setAccessible(true);
                    return method.invoke(target);
                } catch (Exception e) {
                    throw new IllegalStateException(e);
                }
            }
        });
    }

    private Method method(Object target, String methodName, Class<?>[] args)
            throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (int i = 0; i < methods.length; i++) {
            if (methods[i].getName().equals("attachment")) {
                methodName = "attachment";
                break;
            }
        }

        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null)
            return buffer;
        else
            return viewed(viewedBuffer);
    }

    public static void main(String[] args) {
    }
}
